package com.zhang.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.app.func.DimAsyncFunction;
import com.zhang.gmall.realtime.beans.OrderWide;
import com.zhang.gmall.realtime.beans.PaymentWide;
import com.zhang.gmall.realtime.beans.ProductStats;
import com.zhang.gmall.realtime.common.GmallConstant;
import com.zhang.gmall.realtime.utils.ClickHouseUtil;
import com.zhang.gmall.realtime.utils.DateTimeUtil;
import com.zhang.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @title: 商品主题
 * @author: zhang
 * @date: 2022/3/15 08:40
 * 需要启动服务
 * zk,kafka,logger.sh,maxwell,hbase,redis,Clickhouse
 * BaseLogApp,BaseDBApp,OrderWideApp,PaymentWideApp,ProductStatsApp
 */
public class ProductStatsApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        //TODO 2.检查点相关设置
/*        env.enableCheckpointing(60 * 1000L, CheckpointingMode.EXACTLY_ONCE);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flink/gmall-ck"));
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
        System.setProperty("HADOOP_USER_NAME", "zhang");*/

        //TODO 3.读取kafka dwd层 dwd_page_log_2022 主题数据，统计点击数和曝光数
        String pageTopic = "dwd_page_log_2022";
        String groupId = "ProductStatsApp1";

        SingleOutputStreamOperator<ProductStats> clickAndDisplayCntDS = env
                .addSource(MyKafkaUtil.getKafkaSource(pageTopic, groupId))
                .process(new ProcessFunction<String, ProductStats>() {
                    @Override
                    public void processElement(String element, ProcessFunction<String, ProductStats>.Context ctx, Collector<ProductStats> out) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(element);
                        JSONObject page = jsonObject.getJSONObject("page");
                        Long ts = jsonObject.getLong("ts");
                        //只要进入商品详情页就认为是一次点击行为
                        if ("good_detail".equals(page.getString("page_id"))) {
                            ProductStats productStats = ProductStats.builder()
                                    .sku_id(page.getLong("item"))
                                    .click_ct(1L)
                                    .ts(jsonObject.getLong("ts"))
                                    .build();
                            out.collect(productStats);
                        }

                        //尝试取出曝光数据
                        JSONArray displays = jsonObject.getJSONArray("displays");
                        if (displays != null && displays.size() > 0) {
                            for (int i = 0; i < displays.size(); i++) {
                                JSONObject display = displays.getJSONObject(i);
                                if ("sku_id".equals(display.getString("item_type"))) {
                                    out.collect(ProductStats.builder()
                                            .sku_id(display.getLong("item"))
                                            .display_ct(1L)
                                            .ts(jsonObject.getLong("ts"))
                                            .build()
                                    );
                                }
                            }
                        }
                    }
                });

        //clickAndDisplayCntDS.print("clickAndDisplayCntDS");
        //TODO 4.读取kafka dwd层 dwd_favor_info 收藏主题数据，收藏数
        String favorTopic = "dwd_favor_info";
        SingleOutputStreamOperator<ProductStats> favorDS = env
                .addSource(MyKafkaUtil.getKafkaSource(favorTopic, groupId))
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        return ProductStats.builder()
                                .sku_id(jsonObject.getLong("sku_id"))
                                .favor_ct(1L)
                                .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                                .build();
                    }
                });

        //favorDS.print("favorDS");
        //TODO 5.读取kafka dwd层 dwd_cart_info 购物车主题数据，加购数
        String cartTopic = "dwd_cart_info";
        SingleOutputStreamOperator<ProductStats> cartDS = env
                .addSource(MyKafkaUtil.getKafkaSource(cartTopic, groupId))
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        return ProductStats.builder()
                                .sku_id(jsonObject.getLong("sku_id"))
                                .cart_ct(1L)
                                .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                                .build();
                    }
                });

        //cartDS.print("cartDS");

        //TODO 6.读取kafka dwm层 dwm_order_wide_2022 订单宽表数据，下单数
        String orderWideTopic = "dwm_order_wide_2022";
        SingleOutputStreamOperator<ProductStats> orderWideDS = env
                .addSource(MyKafkaUtil.getKafkaSource(orderWideTopic, groupId))
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        OrderWide orderWide = JSON.parseObject(value, OrderWide.class);
                        return ProductStats.builder()
                                .sku_id(orderWide.getSku_id())
                                .order_sku_num(orderWide.getSku_num())
                                .order_amount(orderWide.getSplit_total_amount())
                                .orderIdSet(new HashSet(Collections.singleton(orderWide.getOrder_id())))
                                .ts(DateTimeUtil.toTs(orderWide.getCreate_time()))
                                .build();
                    }
                });

        //orderWideDS.print("orderWideDS");
        //TODO 7.读取kafka dwm层 dwm_payment_wide_2022 支付宽表数据，支付数
        String paymentWideTopic = "dwm_payment_wide_2022";

        SingleOutputStreamOperator<ProductStats> paymentDS = env
                .addSource(MyKafkaUtil.getKafkaSource(paymentWideTopic, groupId))
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        PaymentWide paymentWide = JSON.parseObject(value, PaymentWide.class);
                        return ProductStats.builder()
                                .sku_id(paymentWide.getSku_id())
                                .payment_amount(paymentWide.getSplit_total_amount())
                                .paidOrderIdSet(new HashSet(Collections.singleton(paymentWide.getOrder_id())))
                                .ts(DateTimeUtil.toTs(paymentWide.getCallback_time()))
                                .build();
                    }
                });
        //paymentDS.print("paymentDS");

        //TODO 8.读取kafka dwd层 dwd_refund_payment 退款主题数据，退单数
        String refundPaymentTopic = "dwd_order_refund_info";
        SingleOutputStreamOperator<ProductStats> refundDS = env
                .addSource(MyKafkaUtil.getKafkaSource(refundPaymentTopic, groupId))
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        return ProductStats.builder()
                                .sku_id(jsonObject.getLong("sku_id"))
                                .refund_amount(jsonObject.getBigDecimal("refund_amount"))
                                .refundOrderIdSet(new HashSet(Collections.singleton(jsonObject.getString("order_id"))))
                                .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                                .build();
                    }
                });

        //TODO 9.读取kafka dwd层 dwd_comment_info 评价主题数据，评价数
        String commentTopic = "dwd_comment_info";
        SingleOutputStreamOperator<ProductStats> commentDS = env
                .addSource(MyKafkaUtil.getKafkaSource(commentTopic, groupId))
                .map(new MapFunction<String, ProductStats>() {
                    @Override
                    public ProductStats map(String value) throws Exception {
                        JSONObject jsonObject = JSON.parseObject(value);
                        String appraise = jsonObject.getString("appraise");
                        long goodCnt = 0L;
                        if (GmallConstant.APPRAISE_GOOD.equals(appraise)) goodCnt = 1L;
                        return ProductStats.builder()
                                .sku_id(jsonObject.getLong("sku_id"))
                                .comment_ct(1L)
                                .good_comment_ct(goodCnt)
                                .ts(DateTimeUtil.toTs(jsonObject.getString("create_time")))
                                .build();
                    }
                });

        //TODO 10.七流join、分配水位线和提取时间戳、分组开窗聚合
        SingleOutputStreamOperator<ProductStats> productStatsNoDimDS = clickAndDisplayCntDS
                .union(
                        favorDS,
                        cartDS,
                        orderWideDS,
                        paymentDS,
                        refundDS,
                        commentDS
                )
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ProductStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<ProductStats>() {
                                    @Override
                                    public long extractTimestamp(ProductStats element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                .keyBy(ProductStats::getSku_id)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<ProductStats>() {
                            @Override
                            public ProductStats reduce(ProductStats value1, ProductStats value2) throws Exception {
                                value1.setClick_ct(value1.getClick_ct() + value2.getClick_ct());
                                value1.setDisplay_ct(value1.getDisplay_ct() + value2.getDisplay_ct());
                                value1.setFavor_ct(value1.getFavor_ct() + value2.getFavor_ct());
                                value1.setCart_ct(value1.getCart_ct() + value2.getCart_ct());
                                value1.setComment_ct(value1.getComment_ct() + value2.getComment_ct());
                                value1.setGood_comment_ct(value1.getGood_comment_ct() + value2.getGood_comment_ct());

                                value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                                value1.setOrder_sku_num(value1.getOrder_sku_num() + value2.getOrder_sku_num());
                                value1.setOrder_amount(value1.getOrder_amount().add(value2.getOrder_amount()));

                                value1.getPaidOrderIdSet().addAll(value2.getPaidOrderIdSet());
                                value1.setPayment_amount(value1.getPayment_amount().add(value2.getPayment_amount()));

                                value1.getRefundOrderIdSet().addAll(value2.getPaidOrderIdSet());
                                value1.setRefund_amount(value1.getRefund_amount().add(value2.getRefund_amount()));

                                return value1;
                            }
                        },
                        new ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>() {
                            @Override
                            public void process(Long aLong, ProcessWindowFunction<ProductStats, ProductStats, Long, TimeWindow>.Context context, Iterable<ProductStats> elements, Collector<ProductStats> out) throws Exception {
                                ProductStats productStats = elements.iterator().next();
                                //补充窗口信息
                                productStats.setStt(DateTimeUtil.toYMDhms(new Date(context.window().getStart())));
                                productStats.setEdt(DateTimeUtil.toYMDhms(new Date(context.window().getEnd())));
                                //设置订单数
                                productStats.setOrder_ct((long)productStats.getOrderIdSet().size());
                                productStats.setPaid_order_ct((long)productStats.getPaidOrderIdSet().size());
                                productStats.setRefund_order_ct((long)productStats.getRefundOrderIdSet().size());
                                out.collect(productStats);
                            }
                        }

                );

        //TODO 11.补充维度信息
        //关联sku相关维度信息
        SingleOutputStreamOperator<ProductStats> productStatsSkuDS = AsyncDataStream
                .unorderedWait(
                        productStatsNoDimDS,
                        new DimAsyncFunction<ProductStats>("DIM_SKU_INFO") {
                            @Override
                            public String getKey(ProductStats input) {
                                return input.getSku_id().toString();
                            }

                            @Override
                            public void join(ProductStats input, JSONObject dimInfo) throws ParseException, Exception {
                                input.setSku_name(dimInfo.getString("SKU_NAME"));
                                input.setSku_price(dimInfo.getBigDecimal("PRICE"));
                                input.setSpu_id(dimInfo.getLong("SPU_ID"));
                                input.setTm_id(dimInfo.getLong("TM_ID"));
                                input.setCategory3_id(dimInfo.getLong("CATEGORY3_ID"));
                            }
                        }, 60, TimeUnit.SECONDS);

        //关联spu相关信息
        SingleOutputStreamOperator<ProductStats> productStatsSkuSpuDS = AsyncDataStream
                .unorderedWait(
                        productStatsSkuDS,
                        new DimAsyncFunction<ProductStats>("DIM_SPU_INFO") {
                            @Override
                            public String getKey(ProductStats input) {
                                return input.getSpu_id().toString();
                            }

                            @Override
                            public void join(ProductStats input, JSONObject dimInfo) throws ParseException, Exception {
                                input.setSpu_name(dimInfo.getString("SPU_NAME"));
                            }
                        }, 60, TimeUnit.SECONDS
                );

        //关联品牌相关信息base_trademark
        SingleOutputStreamOperator<ProductStats> productStatsSkuSpuTmDS = AsyncDataStream
                .unorderedWait(
                        productStatsSkuSpuDS,
                        new DimAsyncFunction<ProductStats>("DIM_BASE_TRADEMARK") {
                            @Override
                            public String getKey(ProductStats input) {
                                return input.getTm_id().toString();
                            }

                            @Override
                            public void join(ProductStats input, JSONObject dimInfo) throws ParseException, Exception {
                                input.setTm_name(dimInfo.getString("TM_NAME"));
                            }
                        }, 60, TimeUnit.SECONDS
                );

        //关联商品分类
        SingleOutputStreamOperator<ProductStats> productStatsWithDim = AsyncDataStream
                .unorderedWait(
                        productStatsSkuSpuTmDS,
                        new DimAsyncFunction<ProductStats>("DIM_BASE_CATEGORY3") {
                            @Override
                            public String getKey(ProductStats input) {
                                return input.getCategory3_id().toString();
                            }

                            @Override
                            public void join(ProductStats input, JSONObject dimInfo) throws ParseException, Exception {
                                input.setCategory3_name(dimInfo.getString("NAME"));
                            }
                        }, 60, TimeUnit.SECONDS
                );
        //TODO 11.将聚合结果写入clickhouse
        productStatsWithDim.print();
        productStatsWithDim
                .addSink(ClickHouseUtil.getSinkFunction(
                        "insert into product_stats_2022 values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
                ));
        //TODO 12.执行任务
        env.execute();
    }
}
